package org.firebug.spring.boot.controller;

import com.rabbitmq.client.*;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Client {
    private Logger logger = LoggerFactory.getLogger(this.getClass());
    private static final String QUEUE = "local.api.queue";
    private static final String QUEUE_1 = "local.api.queue.1";
    private static final String QUEUE_2 = "local.api.queue.2";
    private static final String EXCHANGE = "local.api.exchange";
    private static final String EXCHANGE_1 = "local.api.exchange.1";
    private static final String EXCHANGE_2 = "local.api.exchange.2";
    private static final String ROUTING = "local.api.routing";
    private static final String ROUTING_1 = "local.api.routing.1";
    private static final String ROUTING_2 = "local.api.routing.2";
    private static final byte[] MESSAGE = (System.currentTimeMillis() + "").getBytes();

    private ConnectionFactory getFactory() {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("api.firebug.org");
        factory.setPort(5672);
        factory.setUsername("admin");
        factory.setPassword("123456");
        factory.setVirtualHost("/local");
        return factory;
    }

    @Test
    public void produce() throws IOException, TimeoutException {
        Connection connection = getFactory().newConnection();
        Channel channel = connection.createChannel();
//        直连交换机：exchange:routingKey:queue: 1:1:1
//        channel.exchangeDelete(EXCHANGE); //删除交换机
//        channel.queueDelete(QUEUE);//删除队列(自行决定)
//        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT); //声明一个direct交换机
//        channel.queueDeclare(QUEUE, true, false, false, null);//声明一个队列
//        channel.queueBind(QUEUE, EXCHANGE, ROUTING, null);//绑定 交换机与队列
//        channel.basicPublish(EXCHANGE, ROUTING, null, MESSAGE); //发送
//        channel.queueUnbind(QUEUE, EXCHANGE, ROUTING);//解绑 交换机与队列(自行决定)
////        直连交换机：exchange:routingKey:queue: 1:1:2
//        channel.exchangeDelete(EXCHANGE); //删除交换机
//        channel.queueDelete(QUEUE_1);//删除队列(自行决定)
//        channel.queueDelete(QUEUE_2);//删除队列(自行决定)
//        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT); //声明一个direct交换机
//        channel.queueDeclare(QUEUE_1, true, false, false, null);//声明一个队列
//        channel.queueDeclare(QUEUE_2, true, false, false, null);//声明一个队列
//        channel.queueBind(QUEUE_1, EXCHANGE, ROUTING, null);//绑定 交换机与队列
//        channel.queueBind(QUEUE_2, EXCHANGE, ROUTING, null);//绑定 交换机与队列
//        channel.basicPublish(EXCHANGE, ROUTING, null, MESSAGE);//发送
//        channel.queueUnbind(QUEUE_1, EXCHANGE, ROUTING);//解绑 交换机与队列(自行决定)
//        channel.queueUnbind(QUEUE_2, EXCHANGE, ROUTING);//解绑 交换机与队列(自行决定)
////        直连交换机：exchange:routingKey:queue: 1:2:2
//        channel.exchangeDelete(EXCHANGE); //删除交换机
//        channel.queueDelete(QUEUE_1);//删除队列(自行决定)
//        channel.queueDelete(QUEUE_2);//删除队列(自行决定)
//        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT); //声明一个direct交换机
//        channel.queueDeclare(QUEUE_1, true, false, false, null);//声明一个队列
//        channel.queueDeclare(QUEUE_2, true, false, false, null);//声明一个队列
//        channel.queueBind(QUEUE_1, EXCHANGE, ROUTING_1, null);//绑定 交换机与队列
//        channel.queueBind(QUEUE_2, EXCHANGE, ROUTING_2, null);//绑定 交换机与队列
//        channel.basicPublish(EXCHANGE, ROUTING_1, null, MESSAGE);//发送
//        channel.basicPublish(EXCHANGE, ROUTING_2, null, MESSAGE);//发送
//        channel.queueUnbind(QUEUE_1, EXCHANGE, ROUTING_1);//解绑 交换机与队列(自行决定)
//        channel.queueUnbind(QUEUE_2, EXCHANGE, ROUTING_1);//解绑 交换机与队列(自行决定)
////        扇型交换机exchange:routingKey:queue: 1:0:2
//        channel.exchangeDelete(EXCHANGE); //删除交换机
//        channel.queueDelete(QUEUE_1);//删除队列(自行决定)
//        channel.queueDelete(QUEUE_2);//删除队列(自行决定)
//        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.FANOUT);// 声明一个fanout交换机
//        channel.queueDeclare(QUEUE_1, true, false, false, null);//声明一个队列
//        channel.queueDeclare(QUEUE_2, true, false, false, null);//声明一个队列
//        channel.queueBind(QUEUE_1, EXCHANGE, "", null);//绑定 交换机与队列
//        channel.queueBind(QUEUE_2, EXCHANGE, "", null);//绑定 交换机与队列
//        channel.basicPublish(EXCHANGE, "", null, MESSAGE);//发送
//        channel.queueUnbind(QUEUE_1, EXCHANGE, ROUTING);//解绑 交换机与队列(自行决定)
//        channel.queueUnbind(QUEUE_2, EXCHANGE, ROUTING);//解绑 交换机与队列(自行决定)
////        主题交换机exchange:routingKey:queue: 1:*#:2
//        channel.exchangeDelete(EXCHANGE); //删除交换机
//        channel.queueDelete(QUEUE_1);//删除队列(自行决定)
//        channel.queueDelete(QUEUE_2);//删除队列(自行决定)
//        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.TOPIC);// 声明一个fanout交换机
//        channel.queueDeclare(QUEUE_1, true, false, false, null);//声明一个队列
//        channel.queueDeclare(QUEUE_2, true, false, false, null);//声明一个队列
//        channel.queueBind(QUEUE_1, EXCHANGE, ROUTING + ".*", null);//绑定 交换机与队列, *: 1个单词
//        channel.queueBind(QUEUE_2, EXCHANGE, ROUTING + ".#", null);//绑定 交换机与队列, #:0+个单词
//        channel.basicPublish(EXCHANGE, ROUTING + ".0", null, MESSAGE);//发送
//        channel.basicPublish(EXCHANGE, ROUTING + ".1", null, MESSAGE);//发送
//        channel.queueUnbind(QUEUE_1, EXCHANGE, ROUTING + ".*");//解绑 交换机与队列(自行决定)
//        channel.queueUnbind(QUEUE_2, EXCHANGE, ROUTING + ".#");//解绑 交换机与队列(自行决定)
//        头交换机exchange:routingKey:queue:
//        channel.exchangeDelete(EXCHANGE); //删除交换机
//        channel.queueDelete(QUEUE_1);//删除队列(自行决定)
//        channel.queueDelete(QUEUE_2);//删除队列(自行决定)
//        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.HEADERS);// 声明一个fanout交换机
//        channel.queueDeclare(QUEUE_1, true, false, false, null);//声明一个队列
//        channel.queueDeclare(QUEUE_2, true, false, false, null);//声明一个队列
//        Map<String, Object> arg1 = new HashMap<>();
//        arg1.put("format", "pdf");
//        arg1.put("type", "report");
//        arg1.put("x-match", "any");//匹配：A || B
//        channel.queueBind(QUEUE_1, EXCHANGE, "", arg1);//绑定 交换机与队列
//        Map<String, Object> arg2 = new HashMap<>();
//        arg2.put("format", "pdf");
//        arg2.put("type", "report");
//        arg2.put("x-match", "all");//匹配：A && B(默认)
//        channel.queueBind(QUEUE_2, EXCHANGE, "", arg2);//绑定 交换机与队列
//        Map<String, Object> headers = new HashMap<>();
//        headers.put("format", "pdf");
//        headers.put("type", "report");
//        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().headers(headers).build();
//        channel.basicPublish(EXCHANGE, "", props, MESSAGE);//发送
//        channel.queueUnbind(QUEUE_1, EXCHANGE, "");//解绑 交换机与队列(自行决定)
//        channel.queueUnbind(QUEUE_2, EXCHANGE, "");//解绑 交换机与队列(自行决定)
////    延时消息,超时丢弃
//        channel.exchangeDelete(EXCHANGE_1); //删除交换机(自行决定)
//        channel.queueDelete(QUEUE_1);//删除队列(自行决定)
//        channel.exchangeDeclare(EXCHANGE_1, BuiltinExchangeType.DIRECT); //声明一个direct交换机
//        channel.queueDeclare(QUEUE_1, true, false, false, null);//声明一个队列
//        channel.queueBind(QUEUE_1, EXCHANGE_1, ROUTING_1, null);//绑定 交换机与队列
//        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().expiration("10000").build(); //延时10秒,超时丢弃
//        channel.basicPublish(EXCHANGE_1, ROUTING_1, props, MESSAGE);
//        channel.queueUnbind(QUEUE_1, EXCHANGE_1, ROUTING_1);//解绑 交换机与队列(自行决定)
////    延时消息队列
//        channel.exchangeDelete(EXCHANGE_1); //删除交换机(自行决定)
//        channel.exchangeDelete(EXCHANGE_2); //删除交换机(自行决定)
//        channel.queueDelete(QUEUE_1);//删除队列(自行决定)
//        channel.queueDelete(QUEUE_2);//删除队列(自行决定)
//        channel.exchangeDeclare(EXCHANGE_1, BuiltinExchangeType.DIRECT); //声明一个direct交换机
//        channel.exchangeDeclare(EXCHANGE_2, BuiltinExchangeType.DIRECT); //声明一个direct交换机
//        Map<String, Object> args = new HashMap<>();
//        args.put("x-expires", 30000); //队列过期时间
//        args.put("x-message-ttl", 10000);//队列上消息过期时间，应小于队列过期时间
//        args.put("x-dead-letter-exchange", EXCHANGE_2);//过期消息转向路由
//        args.put("x-dead-letter-routing-key", ROUTING_2);//过期消息转向路由相匹配
//        channel.queueDeclare(QUEUE_1, true, false, false, args); //声明一个延时队列
//        channel.queueDeclare(QUEUE_2, true, false, false, null);//声明一个普通队列
//        channel.queueBind(QUEUE_1, EXCHANGE_1, ROUTING_1, null);//绑定 交换机与队列
//        channel.queueBind(QUEUE_2, EXCHANGE_2, ROUTING_2, null);//绑定 交换机与队列
//        channel.basicPublish(EXCHANGE_1, ROUTING_1, null, MESSAGE);
//        channel.queueUnbind(QUEUE_1, EXCHANGE_1, ROUTING_1);//解绑 交换机与队列(自行决定)
//        channel.queueUnbind(QUEUE_2, EXCHANGE_1, ROUTING_2);//解绑 交换机与队列(自行决定)
////    延时消息队列+延时消息
//        channel.exchangeDelete(EXCHANGE_1); //删除交换机(自行决定)
//        channel.exchangeDelete(EXCHANGE_2); //删除交换机(自行决定)
//        channel.queueDelete(QUEUE_1);//删除队列(自行决定)
//        channel.queueDelete(QUEUE_2);//删除队列(自行决定)
//        channel.exchangeDeclare(EXCHANGE_1, BuiltinExchangeType.DIRECT); //声明一个direct交换机
//        channel.exchangeDeclare(EXCHANGE_2, BuiltinExchangeType.DIRECT); //声明一个direct交换机
//        Map<String, Object> args = new HashMap<>();
//        args.put("x-expires", 30000); //队列过期时间
//        args.put("x-message-ttl", 10000);//队列上消息过期时间，应小于队列过期时间
//        args.put("x-dead-letter-exchange", EXCHANGE_2);//过期消息转向路由
//        args.put("x-dead-letter-routing-key", ROUTING_2);//过期消息转向路由相匹配
//        channel.queueDeclare(QUEUE_1, true, false, false, args); //声明一个延时队列
//        channel.queueDeclare(QUEUE_2, true, false, false, null);//声明一个普通队列
//        channel.queueBind(QUEUE_1, EXCHANGE_1, ROUTING_1, null);//绑定 交换机与队列
//        channel.queueBind(QUEUE_2, EXCHANGE_2, ROUTING_2, null);//绑定 交换机与队列
//        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().expiration("5000").build(); //延时5秒,超时丢弃
//        channel.basicPublish(EXCHANGE_1, ROUTING_1, props, MESSAGE);
//        channel.queueUnbind(QUEUE_1, EXCHANGE_1, ROUTING_1);//解绑 交换机与队列(自行决定)
//        channel.queueUnbind(QUEUE_2, EXCHANGE_1, ROUTING_2);//解绑 交换机与队列(自行决定)
        channel.close();
        connection.close();
    }

    @Test
    public void consume() throws IOException, TimeoutException, InterruptedException {
        Connection connection = getFactory().newConnection();
        Channel channel = connection.createChannel();
        channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT); //声明一个direct交换机
        channel.queueBind(QUEUE, EXCHANGE, ROUTING); //绑定 交换机与队列

        // 默认消费者实现
        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
                    throws IOException {
                String message = new String(body, "UTF-8");
                logger.info("consumerTag={}, routingKey={}, body={}", consumerTag, envelope.getRoutingKey(), message);
                channel.basicAck(envelope.getDeliveryTag(), false);// 确认单条消息
//                channel.basicAck(envelope.getDeliveryTag(),true);// 确认批量消息
//                channel.basicReject(envelope.getDeliveryTag(),false); //拒绝单条消息
//                channel.basicNack(envelope.getDeliveryTag(),true, false);//拒绝批量消息
            }
        };

        channel.basicConsume(QUEUE, false, consumer);// 非自动确认接收消息


        Thread.sleep(10000L);
        channel.queueUnbind(QUEUE, EXCHANGE, ROUTING);//解绑 交换机与队列(自行决定)
        channel.close();
        connection.close();
    }
}
